/*-------------------------------------------------------------------------
 *
 * proto.c
 *		逻辑复制协议函数
 *
 * Copyright (c) 2015-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *		src/backend/replication/logical/proto.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/sysattr.h"
#include "catalog/pg_namespace.h"
#include "catalog/pg_type.h"
#include "libpq/pqformat.h"
#include "replication/logicalproto.h"
#include "utils/lsyscache.h"
#include "utils/syscache.h"

/*
 * 协议消息标志。
 */
#define LOGICALREP_IS_REPLICA_IDENTITY 1

#define MESSAGE_TRANSACTIONAL (1<<0)
#define TRUNCATE_CASCADE		(1<<0)
#define TRUNCATE_RESTART_SEQS	(1<<1)

static void fc_logicalrep_write_attrs(StringInfo fc_out, Relation fc_rel,
								   Bitmapset *fc_columns);
static void fc_logicalrep_write_tuple(StringInfo fc_out, Relation fc_rel,
								   TupleTableSlot *fc_slot,
								   bool fc_binary, Bitmapset *fc_columns);
static void fc_logicalrep_read_attrs(StringInfo fc_in, LogicalRepRelation *fc_rel);
static void fc_logicalrep_read_tuple(StringInfo fc_in, LogicalRepTupleData *fc_tuple);

static void fc_logicalrep_write_namespace(StringInfo fc_out, Oid fc_nspid);
static const char *fc_logicalrep_read_namespace(StringInfo fc_in);

/*
 * 检查一列是否被列清单覆盖。
 *
 * 需要小心处理 NULL，因为它被视为覆盖
 * 所有列的列清单。
 */
static bool fc_column_in_column_list(int fc_attnum, Bitmapset *fc_columns)
{
	return (fc_columns == NULL || bms_is_member(fc_attnum, fc_columns));
}


/*
 * 将 BEGIN 写入输出流。
 */
void logicalrep_write_begin(StringInfo fc_out, ReorderBufferTXN *fc_txn)
{
	pq_sendbyte(fc_out, LOGICAL_REP_MSG_BEGIN);

	/* 固定字段 */
	pq_sendint64(fc_out, fc_txn->final_lsn);
	pq_sendint64(fc_out, fc_txn->xact_time.commit_time);
	pq_sendint32(fc_out, fc_txn->xid);
}

/*
 * 从流中读取事务 BEGIN。
 */
void logicalrep_read_begin(StringInfo fc_in, LogicalRepBeginData *fc_begin_data)
{
	/* 读取字段 */
	fc_begin_data->final_lsn = pq_getmsgint64(fc_in);
	if (fc_begin_data->final_lsn == InvalidXLogRecPtr)
		elog(ERROR, "final_lsn not set in begin message");
	fc_begin_data->committime = pq_getmsgint64(fc_in);
	fc_begin_data->xid = pq_getmsgint(fc_in, 4);
}


/*
 * 将 COMMIT 写入输出流。
 */
void logicalrep_write_commit(StringInfo fc_out, ReorderBufferTXN *fc_txn,
						XLogRecPtr fc_commit_lsn)
{
	uint8		fc_flags = 0;

	pq_sendbyte(fc_out, LOGICAL_REP_MSG_COMMIT);

	/* 发送标志字段（暂时未使用） */
	pq_sendbyte(fc_out, fc_flags);

	/* 发送字段 */
	pq_sendint64(fc_out, fc_commit_lsn);
	pq_sendint64(fc_out, fc_txn->end_lsn);
	pq_sendint64(fc_out, fc_txn->xact_time.commit_time);
}

/*
 * 从流中读取事务 COMMIT。
 */
void logicalrep_read_commit(StringInfo fc_in, LogicalRepCommitData *fc_commit_data)
{
	/* 读取标志（暂时未使用） */
	uint8		fc_flags = pq_getmsgbyte(fc_in);

	if (fc_flags != 0)
		elog(ERROR, "unrecognized flags %u in commit message", fc_flags);

	/* 读取字段 */
	fc_commit_data->commit_lsn = pq_getmsgint64(fc_in);
	fc_commit_data->end_lsn = pq_getmsgint64(fc_in);
	fc_commit_data->committime = pq_getmsgint64(fc_in);
}

/*
 * 将 BEGIN PREPARE 写入输出流。
 */
void logicalrep_write_begin_prepare(StringInfo fc_out, ReorderBufferTXN *fc_txn)
{
	pq_sendbyte(fc_out, LOGICAL_REP_MSG_BEGIN_PREPARE);

	/* 固定字段 */
	pq_sendint64(fc_out, fc_txn->final_lsn);
	pq_sendint64(fc_out, fc_txn->end_lsn);
	pq_sendint64(fc_out, fc_txn->xact_time.prepare_time);
	pq_sendint32(fc_out, fc_txn->xid);

	/* 发送 gid */
	pq_sendstring(fc_out, fc_txn->gid);
}

/*
 * 从流中读取事务 BEGIN PREPARE。
 */
void logicalrep_read_begin_prepare(StringInfo fc_in, LogicalRepPreparedTxnData *fc_begin_data)
{
	/* 读取字段 */
	fc_begin_data->prepare_lsn = pq_getmsgint64(fc_in);
	if (fc_begin_data->prepare_lsn == InvalidXLogRecPtr)
		elog(ERROR, "prepare_lsn not set in begin prepare message");
	fc_begin_data->end_lsn = pq_getmsgint64(fc_in);
	if (fc_begin_data->end_lsn == InvalidXLogRecPtr)
		elog(ERROR, "end_lsn not set in begin prepare message");
	fc_begin_data->prepare_time = pq_getmsgint64(fc_in);
	fc_begin_data->xid = pq_getmsgint(fc_in, 4);

	/* 读取 gid（将其复制到预分配缓冲区） */
	strlcpy(fc_begin_data->gid, pq_getmsgstring(fc_in), sizeof(fc_begin_data->gid));
}

/*
 * logicalrep_write_prepare 和
 * logicalrep_write_stream_prepare 的核心功能。
 */
static void fc_logicalrep_write_prepare_common(StringInfo fc_out, LogicalRepMsgType fc_type,
								ReorderBufferTXN *fc_txn, XLogRecPtr fc_prepare_lsn)
{
	uint8		fc_flags = 0;

	pq_sendbyte(fc_out, fc_type);

	/*
	 * 这只能在两阶段提交事务中发生，
	 * 在这种情况下，我们期望有一个有效的 GID。
	 */
	Assert(fc_txn->gid != NULL);
	Assert(rbtxn_prepared(fc_txn));
	Assert(TransactionIdIsValid(fc_txn->xid));

	/* 发送标志字段 */
	pq_sendbyte(fc_out, fc_flags);

	/* 发送字段 */
	pq_sendint64(fc_out, fc_prepare_lsn);
	pq_sendint64(fc_out, fc_txn->end_lsn);
	pq_sendint64(fc_out, fc_txn->xact_time.prepare_time);
	pq_sendint32(fc_out, fc_txn->xid);

	/* 发送 gid */
	pq_sendstring(fc_out, fc_txn->gid);
}

/*
 * 将 PREPARE 写入输出流。
 */
void logicalrep_write_prepare(StringInfo fc_out, ReorderBufferTXN *fc_txn,
						 XLogRecPtr fc_prepare_lsn)
{
	fc_logicalrep_write_prepare_common(fc_out, LOGICAL_REP_MSG_PREPARE,
									fc_txn, fc_prepare_lsn);
}

/*
 * logicalrep_read_prepare 和
 * logicalrep_read_stream_prepare 的核心功能。
 */
static void fc_logicalrep_read_prepare_common(StringInfo fc_in, char *fc_msgtype,
							   LogicalRepPreparedTxnData *fc_prepare_data)
{
	/* 读取标志 */
	uint8		fc_flags = pq_getmsgbyte(fc_in);

	if (fc_flags != 0)
		elog(ERROR, "unrecognized flags %u in %s message", fc_flags, fc_msgtype);

	/* 读取字段 */
	fc_prepare_data->prepare_lsn = pq_getmsgint64(fc_in);
	if (fc_prepare_data->prepare_lsn == InvalidXLogRecPtr)
		elog(ERROR, "prepare_lsn is not set in %s message", fc_msgtype);
	fc_prepare_data->end_lsn = pq_getmsgint64(fc_in);
	if (fc_prepare_data->end_lsn == InvalidXLogRecPtr)
		elog(ERROR, "end_lsn is not set in %s message", fc_msgtype);
	fc_prepare_data->prepare_time = pq_getmsgint64(fc_in);
	fc_prepare_data->xid = pq_getmsgint(fc_in, 4);
	if (fc_prepare_data->xid == InvalidTransactionId)
		elog(ERROR, "invalid two-phase transaction ID in %s message", fc_msgtype);

	/* 读取 gid（将其复制到预分配缓冲区） */
	strlcpy(fc_prepare_data->gid, pq_getmsgstring(fc_in), sizeof(fc_prepare_data->gid));
}

/*
 * 从流中读取事务 PREPARE。
 */
void logicalrep_read_prepare(StringInfo fc_in, LogicalRepPreparedTxnData *fc_prepare_data)
{
	fc_logicalrep_read_prepare_common(fc_in, "prepare", fc_prepare_data);
}

/*
 * 将 COMMIT PREPARED 写入输出流。
 */
void logicalrep_write_commit_prepared(StringInfo fc_out, ReorderBufferTXN *fc_txn,
								 XLogRecPtr fc_commit_lsn)
{
	uint8		fc_flags = 0;

	pq_sendbyte(fc_out, LOGICAL_REP_MSG_COMMIT_PREPARED);

	/*
	 * 这只能在两阶段提交事务中发生，
	 * 在这种情况下，我们期望有一个有效的 GID。
	 */
	Assert(fc_txn->gid != NULL);

	/* 发送标志字段 */
	pq_sendbyte(fc_out, fc_flags);

	/* 发送字段 */
	pq_sendint64(fc_out, fc_commit_lsn);
	pq_sendint64(fc_out, fc_txn->end_lsn);
	pq_sendint64(fc_out, fc_txn->xact_time.commit_time);
	pq_sendint32(fc_out, fc_txn->xid);

	/* 发送 gid */
	pq_sendstring(fc_out, fc_txn->gid);
}

/*
 * 从流中读取事务 COMMIT PREPARED。
 */
void logicalrep_read_commit_prepared(StringInfo fc_in, LogicalRepCommitPreparedTxnData *fc_prepare_data)
{
	/* 读取标志 */
	uint8		fc_flags = pq_getmsgbyte(fc_in);

	if (fc_flags != 0)
		elog(ERROR, "unrecognized flags %u in commit prepared message", fc_flags);

	/* 读取字段 */
	fc_prepare_data->commit_lsn = pq_getmsgint64(fc_in);
	if (fc_prepare_data->commit_lsn == InvalidXLogRecPtr)
		elog(ERROR, "commit_lsn is not set in commit prepared message");
	fc_prepare_data->end_lsn = pq_getmsgint64(fc_in);
	if (fc_prepare_data->end_lsn == InvalidXLogRecPtr)
		elog(ERROR, "end_lsn is not set in commit prepared message");
	fc_prepare_data->commit_time = pq_getmsgint64(fc_in);
	fc_prepare_data->xid = pq_getmsgint(fc_in, 4);

	/* 读取 gid（将其复制到预分配缓冲区） */
	strlcpy(fc_prepare_data->gid, pq_getmsgstring(fc_in), sizeof(fc_prepare_data->gid));
}

/*
 * 将 ROLLBACK PREPARED 写入输出流。
 */
void logicalrep_write_rollback_prepared(StringInfo fc_out, ReorderBufferTXN *fc_txn,
								   XLogRecPtr fc_prepare_end_lsn,
								   TimestampTz fc_prepare_time)
{
	uint8		fc_flags = 0;

	pq_sendbyte(fc_out, LOGICAL_REP_MSG_ROLLBACK_PREPARED);

	/*
	 * 这只能在两阶段提交事务中发生，
	 * 在这种情况下，我们期望有一个有效的 GID。
	 */
	Assert(fc_txn->gid != NULL);

	/* 发送标志字段 */
	pq_sendbyte(fc_out, fc_flags);

	/* 发送字段 */
	pq_sendint64(fc_out, fc_prepare_end_lsn);
	pq_sendint64(fc_out, fc_txn->end_lsn);
	pq_sendint64(fc_out, fc_prepare_time);
	pq_sendint64(fc_out, fc_txn->xact_time.commit_time);
	pq_sendint32(fc_out, fc_txn->xid);

	/* 发送 gid */
	pq_sendstring(fc_out, fc_txn->gid);
}

/*
 * 从流中读取事务 ROLLBACK PREPARED。
 */
void logicalrep_read_rollback_prepared(StringInfo fc_in,
								  LogicalRepRollbackPreparedTxnData *fc_rollback_data)
{
	/* 读取标志 */
	uint8		fc_flags = pq_getmsgbyte(fc_in);

	if (fc_flags != 0)
		elog(ERROR, "unrecognized flags %u in rollback prepared message", fc_flags);

	/* 读取字段 */
	fc_rollback_data->prepare_end_lsn = pq_getmsgint64(fc_in);
	if (fc_rollback_data->prepare_end_lsn == InvalidXLogRecPtr)
		elog(ERROR, "prepare_end_lsn is not set in rollback prepared message");
	fc_rollback_data->rollback_end_lsn = pq_getmsgint64(fc_in);
	if (fc_rollback_data->rollback_end_lsn == InvalidXLogRecPtr)
		elog(ERROR, "rollback_end_lsn is not set in rollback prepared message");
	fc_rollback_data->prepare_time = pq_getmsgint64(fc_in);
	fc_rollback_data->rollback_time = pq_getmsgint64(fc_in);
	fc_rollback_data->xid = pq_getmsgint(fc_in, 4);

	/* 读取 gid（将其复制到预分配缓冲区） */
	strlcpy(fc_rollback_data->gid, pq_getmsgstring(fc_in), sizeof(fc_rollback_data->gid));
}

/*
 * 将 STREAM PREPARE 写入输出流。
 */
void logicalrep_write_stream_prepare(StringInfo fc_out,
								ReorderBufferTXN *fc_txn,
								XLogRecPtr fc_prepare_lsn)
{
	fc_logicalrep_write_prepare_common(fc_out, LOGICAL_REP_MSG_STREAM_PREPARE,
									fc_txn, fc_prepare_lsn);
}

/*
 * 从流中读取 STREAM PREPARE。
 */
void logicalrep_read_stream_prepare(StringInfo fc_in, LogicalRepPreparedTxnData *fc_prepare_data)
{
	fc_logicalrep_read_prepare_common(fc_in, "stream prepare", fc_prepare_data);
}

/*
 * 将 ORIGIN 写入输出流。
 */
void logicalrep_write_origin(StringInfo fc_out, const char *fc_origin,
						XLogRecPtr fc_origin_lsn)
{
	pq_sendbyte(fc_out, LOGICAL_REP_MSG_ORIGIN);

	/* 固定字段 */
	pq_sendint64(fc_out, fc_origin_lsn);

	/* 原字符串 */
	pq_sendstring(fc_out, fc_origin);
}

/*
 * 从输出流中读取 ORIGIN。
 */
char * logicalrep_read_origin(StringInfo fc_in, XLogRecPtr *fc_origin_lsn)
{
	/* 固定字段 */
	*fc_origin_lsn = pq_getmsgint64(fc_in);

	/* 返回原值 */
	return pstrdup(pq_getmsgstring(fc_in));
}

/*
 * 将 INSERT 写入输出流。
 */
void logicalrep_write_insert(StringInfo fc_out, TransactionId fc_xid, Relation fc_rel,
						TupleTableSlot *fc_newslot, bool fc_binary, Bitmapset *fc_columns)
{
	pq_sendbyte(fc_out, LOGICAL_REP_MSG_INSERT);

	/* 事务 ID（如果无效，则我们不会进行流处理） */
	if (TransactionIdIsValid(fc_xid))
		pq_sendint32(fc_out, fc_xid);

	/* 使用 Oid 作为关系标识符 */
	pq_sendint32(fc_out, RelationGetRelid(fc_rel));

	pq_sendbyte(fc_out, 'N');		/* 新元组后续跟随 */
	fc_logicalrep_write_tuple(fc_out, fc_rel, fc_newslot, fc_binary, fc_columns);
}

/*
 * 从流中读取 INSERT。
 *
 * 填充新元组。
 */
LogicalRepRelId logicalrep_read_insert(StringInfo fc_in, LogicalRepTupleData *fc_newtup)
{
	char		fc_action;
	LogicalRepRelId fc_relid;

	/* 读取关系 ID */
	fc_relid = pq_getmsgint(fc_in, 4);

	fc_action = pq_getmsgbyte(fc_in);
	if (fc_action != 'N')
		elog(ERROR, "expected new tuple but got %d",
			 fc_action);

	fc_logicalrep_read_tuple(fc_in, fc_newtup);

	return fc_relid;
}

/*
 * 将 UPDATE 写入输出流。
 */
void logicalrep_write_update(StringInfo fc_out, TransactionId fc_xid, Relation fc_rel,
						TupleTableSlot *fc_oldslot, TupleTableSlot *fc_newslot,
						bool fc_binary, Bitmapset *fc_columns)
{
	pq_sendbyte(fc_out, LOGICAL_REP_MSG_UPDATE);

	Assert(fc_rel->rd_rel->relreplident == REPLICA_IDENTITY_DEFAULT ||
		   fc_rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL ||
		   fc_rel->rd_rel->relreplident == REPLICA_IDENTITY_INDEX);

	/* 事务 ID（如果无效，则我们不会进行流处理） */
	if (TransactionIdIsValid(fc_xid))
		pq_sendint32(fc_out, fc_xid);

	/* 使用 Oid 作为关系标识符 */
	pq_sendint32(fc_out, RelationGetRelid(fc_rel));

	if (fc_oldslot != NULL)
	{
		if (fc_rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL)
			pq_sendbyte(fc_out, 'O');	/* 旧元组后续内容 */
		else
			pq_sendbyte(fc_out, 'K');	/* 旧键后续内容 */
		fc_logicalrep_write_tuple(fc_out, fc_rel, fc_oldslot, fc_binary, fc_columns);
	}

	pq_sendbyte(fc_out, 'N');		/* 新元组后续跟随 */
	fc_logicalrep_write_tuple(fc_out, fc_rel, fc_newslot, fc_binary, fc_columns);
}

/*
 * 从流中读取 UPDATE。
 */
LogicalRepRelId logicalrep_read_update(StringInfo fc_in, bool *fc_has_oldtuple,
					   LogicalRepTupleData *fc_oldtup,
					   LogicalRepTupleData *fc_newtup)
{
	char		fc_action;
	LogicalRepRelId fc_relid;

	/* 读取关系 ID */
	fc_relid = pq_getmsgint(fc_in, 4);

	/* 读取并验证操作 */
	fc_action = pq_getmsgbyte(fc_in);
	if (fc_action != 'K' && fc_action != 'O' && fc_action != 'N')
		elog(ERROR, "expected action 'N', 'O' or 'K', got %c",
			 fc_action);

	/* 检查旧元组 */
	if (fc_action == 'K' || fc_action == 'O')
	{
		fc_logicalrep_read_tuple(fc_in, fc_oldtup);
		*fc_has_oldtuple = true;

		fc_action = pq_getmsgbyte(fc_in);
	}
	else
		*fc_has_oldtuple = false;

	/* 检查新元组 */
	if (fc_action != 'N')
		elog(ERROR, "expected action 'N', got %c",
			 fc_action);

	fc_logicalrep_read_tuple(fc_in, fc_newtup);

	return fc_relid;
}

/*
 * 将 DELETE 写入输出流。
 */
void logicalrep_write_delete(StringInfo fc_out, TransactionId fc_xid, Relation fc_rel,
						TupleTableSlot *fc_oldslot, bool fc_binary,
						Bitmapset *fc_columns)
{
	Assert(fc_rel->rd_rel->relreplident == REPLICA_IDENTITY_DEFAULT ||
		   fc_rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL ||
		   fc_rel->rd_rel->relreplident == REPLICA_IDENTITY_INDEX);

	pq_sendbyte(fc_out, LOGICAL_REP_MSG_DELETE);

	/* 事务 ID（如果无效，则我们不会进行流处理） */
	if (TransactionIdIsValid(fc_xid))
		pq_sendint32(fc_out, fc_xid);

	/* 使用 Oid 作为关系标识符 */
	pq_sendint32(fc_out, RelationGetRelid(fc_rel));

	if (fc_rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL)
		pq_sendbyte(fc_out, 'O');	/* 旧元组后续内容 */
	else
		pq_sendbyte(fc_out, 'K');	/* 旧键后续内容 */

	fc_logicalrep_write_tuple(fc_out, fc_rel, fc_oldslot, fc_binary, fc_columns);
}

/*
 * 从流中读取 DELETE。
 *
 * 填充旧元组。
 */
LogicalRepRelId logicalrep_read_delete(StringInfo fc_in, LogicalRepTupleData *fc_oldtup)
{
	char		fc_action;
	LogicalRepRelId fc_relid;

	/* 读取关系 ID */
	fc_relid = pq_getmsgint(fc_in, 4);

	/* 读取并验证操作 */
	fc_action = pq_getmsgbyte(fc_in);
	if (fc_action != 'K' && fc_action != 'O')
		elog(ERROR, "expected action 'O' or 'K', got %c", fc_action);

	fc_logicalrep_read_tuple(fc_in, fc_oldtup);

	return fc_relid;
}

/*
 * 将 TRUNCATE 写入输出流。
 */
void logicalrep_write_truncate(StringInfo fc_out,
						  TransactionId fc_xid,
						  int fc_nrelids,
						  Oid fc_relids[],
						  bool fc_cascade, bool fc_restart_seqs)
{
	int			fc_i;
	uint8		fc_flags = 0;

	pq_sendbyte(fc_out, LOGICAL_REP_MSG_TRUNCATE);

	/* 事务 ID（如果无效，则我们不会进行流处理） */
	if (TransactionIdIsValid(fc_xid))
		pq_sendint32(fc_out, fc_xid);

	pq_sendint32(fc_out, fc_nrelids);

	/* 编码并发送截断标志 */
	if (fc_cascade)
		fc_flags |= TRUNCATE_CASCADE;
	if (fc_restart_seqs)
		fc_flags |= TRUNCATE_RESTART_SEQS;
	pq_sendint8(fc_out, fc_flags);

	for (fc_i = 0; fc_i < fc_nrelids; fc_i++)
		pq_sendint32(fc_out, fc_relids[fc_i]);
}

/*
 * 从流中读取 TRUNCATE。
 */
List * logicalrep_read_truncate(StringInfo fc_in,
						 bool *fc_cascade, bool *fc_restart_seqs)
{
	int			fc_i;
	int			fc_nrelids;
	List	   *fc_relids = NIL;
	uint8		fc_flags;

	fc_nrelids = pq_getmsgint(fc_in, 4);

	/* 读取并解码截断标志 */
	fc_flags = pq_getmsgint(fc_in, 1);
	*fc_cascade = (fc_flags & TRUNCATE_CASCADE) > 0;
	*fc_restart_seqs = (fc_flags & TRUNCATE_RESTART_SEQS) > 0;

	for (fc_i = 0; fc_i < fc_nrelids; fc_i++)
		fc_relids = lappend_oid(fc_relids, pq_getmsgint(fc_in, 4));

	return fc_relids;
}

/*
 * 将 MESSAGE 写入流
 */
void logicalrep_write_message(StringInfo fc_out, TransactionId fc_xid, XLogRecPtr fc_lsn,
						 bool fc_transactional, const char *fc_prefix, Size fc_sz,
						 const char *fc_message)
{
	uint8		fc_flags = 0;

	pq_sendbyte(fc_out, LOGICAL_REP_MSG_MESSAGE);

	/* 编码并发送消息标志 */
	if (fc_transactional)
		fc_flags |= MESSAGE_TRANSACTIONAL;

	/* 事务 ID（如果无效，则我们不会进行流处理） */
	if (TransactionIdIsValid(fc_xid))
		pq_sendint32(fc_out, fc_xid);

	pq_sendint8(fc_out, fc_flags);
	pq_sendint64(fc_out, fc_lsn);
	pq_sendstring(fc_out, fc_prefix);
	pq_sendint32(fc_out, fc_sz);
	pq_sendbytes(fc_out, fc_message, fc_sz);
}

/*
 * 将关系描述写入输出流。
 */
void logicalrep_write_rel(StringInfo fc_out, TransactionId fc_xid, Relation fc_rel,
					 Bitmapset *fc_columns)
{
	char	   *fc_relname;

	pq_sendbyte(fc_out, LOGICAL_REP_MSG_RELATION);

	/* 事务 ID（如果无效，则我们不会进行流处理） */
	if (TransactionIdIsValid(fc_xid))
		pq_sendint32(fc_out, fc_xid);

	/* 使用 Oid 作为关系标识符 */
	pq_sendint32(fc_out, RelationGetRelid(fc_rel));

	/* 发送限定关系名称 */
	fc_logicalrep_write_namespace(fc_out, RelationGetNamespace(fc_rel));
	fc_relname = RelationGetRelationName(fc_rel);
	pq_sendstring(fc_out, fc_relname);

	/* 发送副本身份 */
	pq_sendbyte(fc_out, fc_rel->rd_rel->relreplident);

	/* 发送属性信息 */
	fc_logicalrep_write_attrs(fc_out, fc_rel, fc_columns);
}

/*
 * 从流中读取关系信息并返回为 LogicalRepRelation。
 */
LogicalRepRelation *
logicalrep_read_rel(StringInfo fc_in)
{
	LogicalRepRelation *fc_rel = palloc(sizeof(LogicalRepRelation));

	fc_rel->remoteid = pq_getmsgint(fc_in, 4);

	/* 从流中读取关系名称 */
	fc_rel->nspname = pstrdup(fc_logicalrep_read_namespace(fc_in));
	fc_rel->relname = pstrdup(pq_getmsgstring(fc_in));

	/* 读取副本身份。 */
	fc_rel->replident = pq_getmsgbyte(fc_in);

	/* 获取属性描述 */
	fc_logicalrep_read_attrs(fc_in, fc_rel);

	return fc_rel;
}

/*
 * 将类型信息写入输出流。
 *
 * 此函数将始终写入基本类型信息。
 */
void logicalrep_write_typ(StringInfo fc_out, TransactionId fc_xid, Oid fc_typoid)
{
	Oid			fc_basetypoid = getBaseType(fc_typoid);
	HeapTuple	fc_tup;
	Form_pg_type fc_typtup;

	pq_sendbyte(fc_out, LOGICAL_REP_MSG_TYPE);

	/* 事务 ID（如果无效，则我们不会进行流处理） */
	if (TransactionIdIsValid(fc_xid))
		pq_sendint32(fc_out, fc_xid);

	fc_tup = SearchSysCache1(TYPEOID, ObjectIdGetDatum(fc_basetypoid));
	if (!HeapTupleIsValid(fc_tup))
		elog(ERROR, "cache lookup failed for type %u", fc_basetypoid);
	fc_typtup = (Form_pg_type) GETSTRUCT(fc_tup);

	/* 使用 Oid 作为关系标识符 */
	pq_sendint32(fc_out, fc_typoid);

	/* 发送限定类型名称 */
	fc_logicalrep_write_namespace(fc_out, fc_typtup->typnamespace);
	pq_sendstring(fc_out, NameStr(fc_typtup->typname));

	ReleaseSysCache(fc_tup);
}

/*
 * 从输出流中读取类型信息。
 */
void logicalrep_read_typ(StringInfo fc_in, LogicalRepTyp *fc_ltyp)
{
	fc_ltyp->remoteid = pq_getmsgint(fc_in, 4);

	/* 从流中读取类型名称 */
	fc_ltyp->nspname = pstrdup(fc_logicalrep_read_namespace(fc_in));
	fc_ltyp->typname = pstrdup(pq_getmsgstring(fc_in));
}

/*
 * 以尽可能高效的格式将元组写入输出流。
 */
static void fc_logicalrep_write_tuple(StringInfo fc_out, Relation fc_rel, TupleTableSlot *fc_slot,
					   bool fc_binary, Bitmapset *fc_columns)
{
	TupleDesc	fc_desc;
	Datum	   *fc_values;
	bool	   *fc_isnull;
	int			fc_i;
	uint16		fc_nliveatts = 0;

	fc_desc = RelationGetDescr(fc_rel);

	for (fc_i = 0; fc_i < fc_desc->natts; fc_i++)
	{
		Form_pg_attribute fc_att = TupleDescAttr(fc_desc, fc_i);

		if (fc_att->attisdropped || fc_att->attgenerated)
			continue;

		if (!fc_column_in_column_list(fc_att->attnum, fc_columns))
			continue;

		fc_nliveatts++;
	}
	pq_sendint16(fc_out, fc_nliveatts);

	slot_getallattrs(fc_slot);
	fc_values = fc_slot->tts_values;
	fc_isnull = fc_slot->tts_isnull;

	/* 写入值 */
	for (fc_i = 0; fc_i < fc_desc->natts; fc_i++)
	{
		HeapTuple	fc_typtup;
		Form_pg_type fc_typclass;
		Form_pg_attribute fc_att = TupleDescAttr(fc_desc, fc_i);

		if (fc_att->attisdropped || fc_att->attgenerated)
			continue;

		if (!fc_column_in_column_list(fc_att->attnum, fc_columns))
			continue;

		if (fc_isnull[fc_i])
		{
			pq_sendbyte(fc_out, LOGICALREP_COLUMN_NULL);
			continue;
		}

		if (fc_att->attlen == -1 && VARATT_IS_EXTERNAL_ONDISK(fc_values[fc_i]))
		{
			/*
			 * 未更改的加压数据。（请注意，我们不承诺检测
			 * 一般性的不变数据；这只是一个简单的检查，以避免
			 * 不必要地发送大型值。）
			 */
			pq_sendbyte(fc_out, LOGICALREP_COLUMN_UNCHANGED);
			continue;
		}

		fc_typtup = SearchSysCache1(TYPEOID, ObjectIdGetDatum(fc_att->atttypid));
		if (!HeapTupleIsValid(fc_typtup))
			elog(ERROR, "cache lookup failed for type %u", fc_att->atttypid);
		fc_typclass = (Form_pg_type) GETSTRUCT(fc_typtup);

		/*
		 * 如果请求并且类型具有合适的发送函数，则以二进制形式发送。
		 */
		if (fc_binary && OidIsValid(fc_typclass->typsend))
		{
			bytea	   *fc_outputbytes;
			int			fc_len;

			pq_sendbyte(fc_out, LOGICALREP_COLUMN_BINARY);
			fc_outputbytes = OidSendFunctionCall(fc_typclass->typsend, fc_values[fc_i]);
			fc_len = VARSIZE(fc_outputbytes) - VARHDRSZ;
			pq_sendint(fc_out, fc_len, 4);	/* 长度 */
			pq_sendbytes(fc_out, VARDATA(fc_outputbytes), fc_len);	/* 数据 */
			pfree(fc_outputbytes);
		}
		else
		{
			char	   *fc_outputstr;

			pq_sendbyte(fc_out, LOGICALREP_COLUMN_TEXT);
			fc_outputstr = OidOutputFunctionCall(fc_typclass->typoutput, fc_values[fc_i]);
			pq_sendcountedtext(fc_out, fc_outputstr, strlen(fc_outputstr), false);
			pfree(fc_outputstr);
		}

		ReleaseSysCache(fc_typtup);
	}
}

/*
 * 从流中以逻辑复制格式读取元组。
 */
static void fc_logicalrep_read_tuple(StringInfo fc_in, LogicalRepTupleData *fc_tuple)
{
	int			fc_i;
	int			fc_natts;

	/* 获取属性数量 */
	fc_natts = pq_getmsgint(fc_in, 2);

	/* 为每列值分配空间；将未使用的 StringInfoDatas 清零 */
	fc_tuple->colvalues = (StringInfoData *) palloc0(fc_natts * sizeof(StringInfoData));
	fc_tuple->colstatus = (char *) palloc(fc_natts * sizeof(char));
	fc_tuple->ncols = fc_natts;

	/* 读取数据 */
	for (fc_i = 0; fc_i < fc_natts; fc_i++)
	{
		char		fc_kind;
		int			fc_len;
		StringInfo	fc_value = &fc_tuple->colvalues[fc_i];

		fc_kind = pq_getmsgbyte(fc_in);
		fc_tuple->colstatus[fc_i] = fc_kind;

		switch (fc_kind)
		{
			case LOGICALREP_COLUMN_NULL:
				/* 没有更多的工作要做 */
				break;
			case LOGICALREP_COLUMN_UNCHANGED:
				/* 我们不接收未更改列的值 */
				break;
			case LOGICALREP_COLUMN_TEXT:
				fc_len = pq_getmsgint(fc_in, 4);	/* 读取长度 */

				/* 和数据 */
				fc_value->data = palloc(fc_len + 1);
				pq_copymsgbytes(fc_in, fc_value->data, fc_len);
				fc_value->data[fc_len] = '\0';
				/* 使 StringInfo 完全有效 */
				fc_value->len = fc_len;
				fc_value->cursor = 0;
				fc_value->maxlen = fc_len;
				break;
			case LOGICALREP_COLUMN_BINARY:
				fc_len = pq_getmsgint(fc_in, 4);	/* 读取长度 */

				/* 和数据 */
				fc_value->data = palloc(fc_len + 1);
				pq_copymsgbytes(fc_in, fc_value->data, fc_len);
				/* 并不严格必要，但符合 StringInfo 实践 */
				fc_value->data[fc_len] = '\0';
				/* 使 StringInfo 完全有效 */
				fc_value->len = fc_len;
				fc_value->cursor = 0;
				fc_value->maxlen = fc_len;
				break;
			default:
				elog(ERROR, "unrecognized data representation type '%c'", fc_kind);
		}
	}
}

/*
 * 将关系属性元数据写入流。
 */
static void fc_logicalrep_write_attrs(StringInfo fc_out, Relation fc_rel, Bitmapset *fc_columns)
{
	TupleDesc	fc_desc;
	int			fc_i;
	uint16		fc_nliveatts = 0;
	Bitmapset  *fc_idattrs = NULL;
	bool		fc_replidentfull;

	fc_desc = RelationGetDescr(fc_rel);

	/* 发送活动属性的数量 */
	for (fc_i = 0; fc_i < fc_desc->natts; fc_i++)
	{
		Form_pg_attribute fc_att = TupleDescAttr(fc_desc, fc_i);

		if (fc_att->attisdropped || fc_att->attgenerated)
			continue;

		if (!fc_column_in_column_list(fc_att->attnum, fc_columns))
			continue;

		fc_nliveatts++;
	}
	pq_sendint16(fc_out, fc_nliveatts);

	/* 获取REPLICATION IDENTITY属性的位图 */
	fc_replidentfull = (fc_rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL);
	if (!fc_replidentfull)
		fc_idattrs = RelationGetIdentityKeyBitmap(fc_rel);

	/* 发送属性 */
	for (fc_i = 0; fc_i < fc_desc->natts; fc_i++)
	{
		Form_pg_attribute fc_att = TupleDescAttr(fc_desc, fc_i);
		uint8		fc_flags = 0;

		if (fc_att->attisdropped || fc_att->attgenerated)
			continue;

		if (!fc_column_in_column_list(fc_att->attnum, fc_columns))
			continue;

		/* REPLICA IDENTITY FULL意味着所有列作为关键部分被发送。 */
		if (fc_replidentfull ||
			bms_is_member(fc_att->attnum - FirstLowInvalidHeapAttributeNumber,
						  fc_idattrs))
			fc_flags |= LOGICALREP_IS_REPLICA_IDENTITY;

		pq_sendbyte(fc_out, fc_flags);

		/* 属性名称 */
		pq_sendstring(fc_out, NameStr(fc_att->attname));

		/* 属性类型id */
		pq_sendint32(fc_out, (int) fc_att->atttypid);

		/* 属性模式 */
		pq_sendint32(fc_out, fc_att->atttypmod);
	}

	bms_free(fc_idattrs);
}

/*
 * 从流中读取关系属性元数据。
 */
static void fc_logicalrep_read_attrs(StringInfo fc_in, LogicalRepRelation *fc_rel)
{
	int			fc_i;
	int			fc_natts;
	char	  **fc_attnames;
	Oid		   *fc_atttyps;
	Bitmapset  *fc_attkeys = NULL;

	fc_natts = pq_getmsgint(fc_in, 2);
	fc_attnames = palloc(fc_natts * sizeof(char *));
	fc_atttyps = palloc(fc_natts * sizeof(Oid));

	/* 读取属性 */
	for (fc_i = 0; fc_i < fc_natts; fc_i++)
	{
		uint8		fc_flags;

		/* 检查副本标识列 */
		fc_flags = pq_getmsgbyte(fc_in);
		if (fc_flags & LOGICALREP_IS_REPLICA_IDENTITY)
			fc_attkeys = bms_add_member(fc_attkeys, fc_i);

		/* 属性名称 */
		fc_attnames[fc_i] = pstrdup(pq_getmsgstring(fc_in));

		/* 属性类型id */
		fc_atttyps[fc_i] = (Oid) pq_getmsgint(fc_in, 4);

		/* 目前我们忽略属性模式 */
		(void) pq_getmsgint(fc_in, 4);
	}

	fc_rel->attnames = fc_attnames;
	fc_rel->atttyps = fc_atttyps;
	fc_rel->attkeys = fc_attkeys;
	fc_rel->natts = fc_natts;
}

/*
 * 写入命名空间名称或空字符串表示pg_catalog（以节省空间）。
 */
static void fc_logicalrep_write_namespace(StringInfo fc_out, Oid fc_nspid)
{
	if (fc_nspid == PG_CATALOG_NAMESPACE)
		pq_sendbyte(fc_out, '\0');
	else
	{
		char	   *fc_nspname = get_namespace_name(fc_nspid);

		if (fc_nspname == NULL)
			elog(ERROR, "cache lookup failed for namespace %u",
				 fc_nspid);

		pq_sendstring(fc_out, fc_nspname);
	}
}

/*
 * 在将空字符串视为pg_catalog的同时读取命名空间名称。
 */
static const char * fc_logicalrep_read_namespace(StringInfo fc_in)
{
	const char *fc_nspname = pq_getmsgstring(fc_in);

	if (fc_nspname[0] == '\0')
		fc_nspname = "pg_catalog";

	return fc_nspname;
}

/*
 * 将启动流消息的信息写入输出流。
 */
void logicalrep_write_stream_start(StringInfo fc_out,
							  TransactionId fc_xid, bool fc_first_segment)
{
	pq_sendbyte(fc_out, LOGICAL_REP_MSG_STREAM_START);

	Assert(TransactionIdIsValid(fc_xid));

	/* 交易ID（我们开始流，所以必须有效） */
	pq_sendint32(fc_out, fc_xid);

	/* 如果这是该xid的第一个流段，则为1 */
	pq_sendbyte(fc_out, fc_first_segment ? 1 : 0);
}

/*
 * 从输出流中读取启动流消息的信息。
 */
TransactionId logicalrep_read_stream_start(StringInfo fc_in, bool *fc_first_segment)
{
	TransactionId fc_xid;

	Assert(fc_first_segment);

	fc_xid = pq_getmsgint(fc_in, 4);
	*fc_first_segment = (pq_getmsgbyte(fc_in) == 1);

	return fc_xid;
}

/*
 * 将停止流消息写入输出流。
 */
void logicalrep_write_stream_stop(StringInfo fc_out)
{
	pq_sendbyte(fc_out, LOGICAL_REP_MSG_STREAM_STOP);
}

/*
 * 将STREAM COMMIT写入输出流。
 */
void logicalrep_write_stream_commit(StringInfo fc_out, ReorderBufferTXN *fc_txn,
							   XLogRecPtr fc_commit_lsn)
{
	uint8		fc_flags = 0;

	pq_sendbyte(fc_out, LOGICAL_REP_MSG_STREAM_COMMIT);

	Assert(TransactionIdIsValid(fc_txn->xid));

	/* 交易ID */
	pq_sendint32(fc_out, fc_txn->xid);

	/* 发送标志字段（暂时未使用） */
	pq_sendbyte(fc_out, fc_flags);

	/* 发送字段 */
	pq_sendint64(fc_out, fc_commit_lsn);
	pq_sendint64(fc_out, fc_txn->end_lsn);
	pq_sendint64(fc_out, fc_txn->xact_time.commit_time);
}

/*
 * 从输出流中读取STREAM COMMIT。
 */
TransactionId logicalrep_read_stream_commit(StringInfo fc_in, LogicalRepCommitData *fc_commit_data)
{
	TransactionId fc_xid;
	uint8		fc_flags;

	fc_xid = pq_getmsgint(fc_in, 4);

	/* 读取标志（暂时未使用） */
	fc_flags = pq_getmsgbyte(fc_in);

	if (fc_flags != 0)
		elog(ERROR, "unrecognized flags %u in commit message", fc_flags);

	/* 读取字段 */
	fc_commit_data->commit_lsn = pq_getmsgint64(fc_in);
	fc_commit_data->end_lsn = pq_getmsgint64(fc_in);
	fc_commit_data->committime = pq_getmsgint64(fc_in);

	return fc_xid;
}

/*
 * 将STREAM ABORT写入输出流。注意，xid和subxid将在
 * 顶级事务中中止时相同。
 */
void logicalrep_write_stream_abort(StringInfo fc_out, TransactionId fc_xid,
							  TransactionId fc_subxid)
{
	pq_sendbyte(fc_out, LOGICAL_REP_MSG_STREAM_ABORT);

	Assert(TransactionIdIsValid(fc_xid) && TransactionIdIsValid(fc_subxid));

	/* 交易ID */
	pq_sendint32(fc_out, fc_xid);
	pq_sendint32(fc_out, fc_subxid);
}

/*
 * 从输出流中读取STREAM ABORT。
 */
void logicalrep_read_stream_abort(StringInfo fc_in, TransactionId *fc_xid,
							 TransactionId *fc_subxid)
{
	Assert(fc_xid && fc_subxid);

	*fc_xid = pq_getmsgint(fc_in, 4);
	*fc_subxid = pq_getmsgint(fc_in, 4);
}

/*
 * 获取代表LogicalRepMsgType的字符串。
 */
const char * logicalrep_message_type(LogicalRepMsgType fc_action)
{
	static char	fc_err_unknown[20];

	switch (fc_action)
	{
		case LOGICAL_REP_MSG_BEGIN:
			return "BEGIN";
		case LOGICAL_REP_MSG_COMMIT:
			return "COMMIT";
		case LOGICAL_REP_MSG_ORIGIN:
			return "ORIGIN";
		case LOGICAL_REP_MSG_INSERT:
			return "INSERT";
		case LOGICAL_REP_MSG_UPDATE:
			return "UPDATE";
		case LOGICAL_REP_MSG_DELETE:
			return "DELETE";
		case LOGICAL_REP_MSG_TRUNCATE:
			return "TRUNCATE";
		case LOGICAL_REP_MSG_RELATION:
			return "RELATION";
		case LOGICAL_REP_MSG_TYPE:
			return "TYPE";
		case LOGICAL_REP_MSG_MESSAGE:
			return "MESSAGE";
		case LOGICAL_REP_MSG_BEGIN_PREPARE:
			return "BEGIN PREPARE";
		case LOGICAL_REP_MSG_PREPARE:
			return "PREPARE";
		case LOGICAL_REP_MSG_COMMIT_PREPARED:
			return "COMMIT PREPARED";
		case LOGICAL_REP_MSG_ROLLBACK_PREPARED:
			return "ROLLBACK PREPARED";
		case LOGICAL_REP_MSG_STREAM_START:
			return "STREAM START";
		case LOGICAL_REP_MSG_STREAM_STOP:
			return "STREAM STOP";
		case LOGICAL_REP_MSG_STREAM_COMMIT:
			return "STREAM COMMIT";
		case LOGICAL_REP_MSG_STREAM_ABORT:
			return "STREAM ABORT";
		case LOGICAL_REP_MSG_STREAM_PREPARE:
			return "STREAM PREPARE";
	}

	/*
	 * 此消息提供了在应用逻辑消息时引发的错误中的上下文。
	 * 因此我们不能在这里抛出错误。返回一个未知的
	 * 指示值，以便仍然报告原始错误。
	 */
	snprintf(fc_err_unknown, sizeof(fc_err_unknown), "??? (%d)", fc_action);

	return fc_err_unknown;
}
